package com.lsx143.realtime.util;

import com.lsx143.realtime.annotation.NoSink;
import com.lsx143.realtime.bean.PaymentInfo;
import com.lsx143.realtime.common.Constants;
import lombok.SneakyThrows;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;

public class SinkUtil {
    public static void main(String[] args) {
        getClickHouseSink("gmall2021", "a", PaymentInfo.class);
    }

    /**
     * 返回clickhouse-sink
     *
     * @param db    数据库名
     * @param table 表名
     * @param clz   POJO
     * @param <T>   POJO的泛型
     * @return clickhouse-sink
     */
    public static <T> SinkFunction<T> getClickHouseSink(String db, String table, Class<T> clz) {
        String driver = Constants.CLICKHOUSE_DRIVER;
        String url = Constants.CLICKHOUSE_URL + "/" + db;
        //提取出POJO的列名和占位符
        List<String> fieldNames = BeanUtil.getFieldNames(clz);
        String fNames = StringUtils.join(fieldNames.toArray(), ",");
        String ph = BeanUtil.getPlaceholder("?", ",", fieldNames.size());
        //拼接sql
        // insert into table(xx,xx,xx)values(?,?...)
        StringBuilder sql = new StringBuilder();
        sql
                .append("insert into ")
                .append(table)
                .append("(")
                .append(fNames)
                .append(")values(")
                .append(ph)
                .append(")");

        System.out.println("执行sql=" + sql.toString());
        return getJDBCSink(driver, url, sql.toString());
    }

    public static <T> SinkFunction<T> getJDBCSink(String driver, String url, String sql) {
        //得到一个jdbcSink
        return JdbcSink
                .sink(
                        sql,
                        new JdbcStatementBuilder<T>() {
                            @SneakyThrows
                            @Override
                            //连接处理
                            public void accept(PreparedStatement ps, T t) throws SQLException {
                                //反射读取类的属性
                                Class<?> clz = t.getClass();
                                Field[] fields = clz.getDeclaredFields();
                                //遍历替换占位符
                                for (int i = 0, index = 1; i < fields.length; i++) {
                                    Field field = fields[i];
                                    // 如果这个field有NoSink 这个注解, 则不要拼接到sql中
                                    NoSink noSink = field.getAnnotation(NoSink.class);
                                    if (noSink == null) {
                                        field.setAccessible(true);
                                        Object value = field.get(t);
                                        ps.setObject(index++, value);
                                    }
                                }
                            }
                        },
                        new JdbcExecutionOptions.Builder()
                                .withBatchIntervalMs(1000)
                                .withBatchSize(1024)
                                .withMaxRetries(3)
                                .build(),
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withDriverName(driver)
                                .withUrl(url)
                                .build()
                );
    }
}
